package com.raven.play.stream.test.scaladsl.tutorial

import akka.{Done, NotUsed}
import akka.actor.ActorSystem
import akka.stream.OverflowStrategy
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}

import scala.concurrent.{Await, Future}
import scala.concurrent.duration.{Duration, DurationInt}

object $08BackPressureBasics extends App {
  implicit val system = ActorSystem("FirstPrinciples")

  val fastSource = Source(1 to 1000)
  val slowSink = Sink.foreach[Int] {
    x =>
      // simulate a long processing
      Thread.sleep(1000)
      println(s"Sink: $x")
  }

  //  fastSource.to(slowSink).run()

  // async 无效, actor 分开, 无法加速单个元素的处理
  //  fastSource.async.to(slowSink).run()
  // backpressure
  val simpleFlow = Flow[Int].map { x =>
    println(s"Incoming: $x")
    x + 1
  }

//  val done = fastSource.async
//    .via(simpleFlow).async
//    .toMat(slowSink)(Keep.right)
//    .run()

  /**
   * reactions to backpressure (in order):
   * - try to slow down if possible
   * - buffer elements until there's more demand
   * - drop down elements from the buffer if it overflows
   * - tear down/kill the whole stream (failure)
   *
   * OverflowStrategy
   * - dropHead => oldest
   * - dropTail => newest
   * - dropBuffer => drop the entire buffer
   * - dropNew => keep the buffer
   * - backpressure => signal
   * - fail
   */
  // TODO: Very Important
  val bufferedFlow = simpleFlow.buffer(10, overflowStrategy = OverflowStrategy.dropHead)
  val done = fastSource.async
    .via(bufferedFlow).async
    .toMat(slowSink)(Keep.right)
    .run()

  /**
   * 1-16: nobody is backpressured
   * 17-26: flow will buffer,flow will start dropping at the rext element
   * 26-1000: flow will always drop the oldest element
   * 991-1000 => 992 - 1001 => sink
   */

  // throttling
  // 2 ele per 1 sec
  fastSource.throttle(2, 1.second).runWith(Sink.foreach(println))

  Await.result(done, Duration.Inf)
  system.terminate()
}
